package org.ala.distributed_transaction_commons.communication.coordinator;

import org.ala.distributed_transaction_commons.communication.Message;
import org.ala.distributed_transaction_commons.logs.LoggerFactory;
import org.ala.distributed_transaction_commons.redis.DistributedTransactionRedisTools;
import org.slf4j.Logger;
import org.springframework.beans.factory.InitializingBean;

import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Transaction;


/**
 * 向协调者publish消息
 *
 * @author 骆毅(ala)
 * @date 2024年2月18日
 */
public class CoordinatorPublisher implements InitializingBean {

	
	static Logger log = LoggerFactory.getLogger("distributed_transaction/coordinator", "publisher");
	
	
	/**
	 * redis工具
	 */
	protected DistributedTransactionRedisTools distributedTransactionRedisTools;
	
	
	@Override
	public void afterPropertiesSet() throws Exception {
		
	}
	
	
	/**
	 * 向协调者发送begin消息
	 */
	public void sendCoordinatorBegin(Pipeline redisTx, String biz, String txId, String pid) {
		String msg = Message.toCoordinatorBegin(biz, txId, pid);
		redisTx.publish(Message.channelCoordinator(txId), msg);
		
		log.info("[CoordinatorPublisher sendCoordinatorBegin] 向协调者发送begin消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	public void sendCoordinatorBegin(String biz, String txId, String pid) {
		String msg = Message.toCoordinatorBegin(biz, txId, pid);
		distributedTransactionRedisTools.withJedis(jedis -> jedis.publish(Message.channelCoordinator(txId), msg));
		
		log.info("[CoordinatorPublisher sendCoordinatorBegin] 向协调者发送begin消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	/**
	 * 向协调者发送can_commit消息
	 */
	public void sendCoordinatorCanCommit(Pipeline redisTx, String biz, String txId, String pid) {
		String msg = Message.toCoordinatorCanCommit(biz, txId, pid);
		redisTx.publish(Message.channelCoordinator(txId), msg);
		
		log.info("[CoordinatorPublisher sendCoordinatorCanCommit] 向协调者发送can_commit消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	public void sendCoordinatorCanCommit(String biz, String txId, String pid) {
		String msg = Message.toCoordinatorCanCommit(biz, txId, pid);
		distributedTransactionRedisTools.withJedis(jedis -> jedis.publish(Message.channelCoordinator(txId), msg));
		
		log.info("[CoordinatorPublisher sendCoordinatorCanCommit] 向协调者发送can_commit消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	
	
	/**
	 * 向协调者发送pre_commit消息
	 */
	public void sendCoordinatorPreCommit(Pipeline redisTx, String biz, String txId, String pid) {
		String msg = Message.toCoordinatorPreCommit(biz, txId, pid);
		redisTx.publish(Message.channelCoordinator(txId), msg);
		
		log.info("[CoordinatorPublisher sendCoordinatorPreCommit] 向协调者发送pre_commit消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	public void sendCoordinatorPreCommit(String biz, String txId, String pid) {
		String msg = Message.toCoordinatorPreCommit(biz, txId, pid);
		distributedTransactionRedisTools.withJedis(jedis -> jedis.publish(Message.channelCoordinator(txId), msg));
		
		log.info("[CoordinatorPublisher sendCoordinatorPreCommit] 向协调者发送pre_commit消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	
	
	/**
	 * 向协调者发送commited消息
	 */
	public void sendCoordinatorCommited(Transaction redisTx, String biz, String txId, String pid) {
		String msg = Message.toCoordinatorCommited(biz, txId, pid);
		redisTx.publish(Message.channelCoordinator(txId), msg);
		
		log.info("[CoordinatorPublisher sendCoordinatorCommited] 向协调者发送commited消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	public void sendCoordinatorCommited(String biz, String txId, String pid) {
		String msg = Message.toCoordinatorCommited(biz, txId, pid);
		distributedTransactionRedisTools.withJedis(jedis -> jedis.publish(Message.channelCoordinator(txId), msg));
		
		log.info("[CoordinatorPublisher sendCoordinatorCommited] 向协调者发送commited消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	
	
	
	/**
	 * 向协调者发送rollback消息
	 */
	public void sendCoordinatorRollback(Pipeline redisTx, String biz, String txId, String pid) {
		String msg = Message.toCoordinatorRollback(biz, txId, pid);
		redisTx.publish(Message.channelCoordinator(txId), msg);
		
		log.info("[CoordinatorPublisher sendCoordinatorRollback] 向协调者发送rollback消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}
	public void sendCoordinatorRollback(String biz, String txId, String pid) {
		String msg = Message.toCoordinatorRollback(biz, txId, pid);
		distributedTransactionRedisTools.withJedis(jedis -> jedis.publish(Message.channelCoordinator(txId), msg));
		
		log.info("[CoordinatorPublisher sendCoordinatorRollback] 向协调者发送rollback消息 biz:" + biz + " txId:" + txId + " pid:" + pid);
	}


	public DistributedTransactionRedisTools getDistributedTransactionRedisTools() {
		return distributedTransactionRedisTools;
	}
	public void setDistributedTransactionRedisTools(DistributedTransactionRedisTools distributedTransactionRedisTools) {
		this.distributedTransactionRedisTools = distributedTransactionRedisTools;
	}
}
